package com.hs.kfk.serializer;

import org.apache.kafka.clients.producer.*;

import java.util.Properties;
import java.util.concurrent.ExecutionException;

/**
 * @Description: 自定义消息序列化
 * @Author 胡尚
 * @Date: 2024/8/7 18:33
 */
public class UserProducer {

     private final static String BOOTSTRAP_SERVER = "192.168.75.61:9092,192.168.75.62:9092,192.168.75.63:9092";
    private final static String TOPIC_NAME = "disTopic";

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        Properties properties = new Properties();
        properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVER);
        properties.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
        // 配置value的序列化类
        properties.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "com.hs.kfk.serializer.UserSerializer");
        Producer<String, User> producer = new KafkaProducer<>(properties);
        for (int i = 0; i < 5; i++) {
            // 构建一条消息，
            User user = new User((long) i, "hushang", 18 + i);
            ProducerRecord<String, User> record = new ProducerRecord<>(TOPIC_NAME, Integer.toString(i), user);
            // 发送
            RecordMetadata recordMetadata = producer.send(record).get();
        }
        producer.close();
    }
}
